package com.anmi.alfred.service.concreteservice;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.anmi.alfred.entity.TransferCaseOperationResult;
import com.anmi.alfred.service.baseservice.IndexService;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkItemResponse;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.List;
import java.util.Map;

@Service
@Slf4j
public class CaseOperationIndexService extends IndexService {

  @Override
  public void index(String indexName, String docId, String jsonStr) {
    super.index(indexName, docId, jsonStr);
  }

  @Override
  public void indexSync(String indexName, String docId, String jsonStr) {
    super.indexSync(indexName, docId, jsonStr);
  }

  @Override
  public void index(String indexName, String docId, Map<String, Object> map) {
    super.index(indexName, docId, map);
  }

  /**
   * 批量索引文档
   *
   * @param indexName 索引名称
   * @param list 数据
   * @return 由BulkRequest返回的response的数量
   */
  public Integer bulkIndex(String indexName, List<TransferCaseOperationResult> list) {
    if (list != null && list.size() > 0) {
      int index = 0;
      BulkRequest bulkRequest = new BulkRequest();
      // 这里添加全局的pipeline，现在是所有的sub request都是统一的，如果有其他的pipeline，sub request 可以override
      bulkRequest.pipeline("handle_out_serial_no_pipeline");
      bulkRequest.timeout(TimeValue.timeValueMinutes(1));
      for (TransferCaseOperationResult temp : list) {
        bulkRequest.add(
            assembleIndexRequest(
                indexName,
                temp.getId().toString(),
                JSON.toJSONString(temp, SerializerFeature.WriteNullStringAsEmpty)));
      }
      try {
        BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
        if (bulkResponse != null) {
          if (bulkResponse.hasFailures()) {
            // 有一个或多个索引失败，现在简单处理，暂时不需要处理failure message
            log.warn("有文档索引失败");
            for (BulkItemResponse bulkItemResponse : bulkResponse) {
              if (bulkItemResponse.isFailed()) {
                BulkItemResponse.Failure failure = bulkItemResponse.getFailure();
                log.warn("文档{}索引失败的具体信息，{}", failure.getId(), failure.getMessage());
              }
            }
          } else {
            for (BulkItemResponse bulkItemResponse : bulkResponse) {
              switch (bulkItemResponse.getOpType()) {
                case INDEX:
                case CREATE:
                case UPDATE:
                  index++;
                  break;
              }
            }
            log.info("所有文档索引成功");
          }
        }
      } catch (IOException e) {
        log.error("批量索引失败,{}", e.getMessage());
      }
      return index;
    }
    return 0;
  }

  private IndexRequest assembleIndexRequest(String indexName, String docId, String jsonStr) {
    IndexRequest indexRequest = new IndexRequest(indexName).id(docId).type("_doc");
    indexRequest.source(jsonStr, XContentType.JSON);
    return indexRequest;
  }
}
